package cn.itcast.kafka.utils;
import cn.itcast.kafka.config.KafkaProducerConfig;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

/**
 * 死信队列工具类，用于发送处理失败的消息
 */
public class DeadLetterQueueUtil {
    // 死信队列主题（建议命名规范：原主题+".dlq"）
    private static final String DEAD_LETTER_TOPIC = "test.dlq";
    private static KafkaProducer<String, String> producer;

    static {
        // 复用生产者配置
        Properties props = KafkaProducerConfig.getProducerProperties();
        producer = new KafkaProducer<>(props);
    }

    /**
     * 发送消息到死信队列
     * @param originalTopic 原主题
     * @param key 消息key
     * @param value 消息内容
     * @param exception 异常信息
     */
    public static void sendToDLQ(String originalTopic, String key, String value, Exception exception) {
        // 构建死信消息，附加原主题和异常信息
        String dlqMessage = String.format(
                "原主题: %s, 异常: %s, 消息内容: %s",
                originalTopic,
                exception.getMessage(),
                value
        );
        ProducerRecord<String, String> record = new ProducerRecord<>(DEAD_LETTER_TOPIC, key, dlqMessage);
        producer.send(record);
        System.out.println("消息已发送到死信队列: " + dlqMessage);
    }

    /**
     * 关闭生产者（程序退出时调用）
     */
    public static void close() {
        if (producer != null) {
            producer.close();
        }
    }
}